package repo

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/star-find-cloud/star-chat/pkg/database"
	"github.com/star-find-cloud/star-chat/pkg/idgen"
	log "github.com/star-find-cloud/star-chat/pkg/logger"
	"github.com/star-find-cloud/star-chat/rule/domain"
	"time"
)

type RuleRepoImpl struct {
	db    database.Database
	cache *database.Redis
}

func NewRoleRepo(db database.Database, cache *database.Redis) *RuleRepoImpl {
	return &RuleRepoImpl{
		db:    db,
		cache: cache,
	}
}

func (r *RuleRepoImpl) Create(ctx context.Context, rule *domain.Rule) (int64, error) {
	sqlStr := "insert into star_chat.public.rules(name, description, status, conversation_type, target_user_role, permission, messagehanding, time_conditions, trigger_conditions, action, create_time) values ($1,  $2, $3, $4, $5, $6, $7, $8, $9, $10, now());"

	result, err := r.db.GetDB().ExecContext(ctx, sqlStr,
		rule.Name,
		rule.Description,
		rule.Status,
		rule.ConversationType,
		rule.TargetUserRole,
		rule.Permission,
		rule.MessageHanding,
		rule.TimeConditions,
		rule.TriggerConditions,
		rule.Action)

	if err != nil {
		log.SQLLogger.Errorf("create rule failed: %v", err)
		return 0, domain.DatabaseCreateError
	}
	id, err := result.LastInsertId()
	if err != nil {
		log.SQLLogger.Errorf("get rule id failed: %v", err)
		return 0, domain.DatabaseGetError
	}

	return id, nil
}

func (r *RuleRepoImpl) MakeRuleCache(ctx context.Context, id int64, rule *domain.Rule) error {
	key := fmt.Sprintf("rule:%d", id)
	jsonStr, err := json.Marshal(rule)
	if err != nil {
		log.CacheLogger.Errorf("marshal rule cache failed: %v", err)
		return domain.MarshalJsonError
	}

	err = r.cache.GetCache().Set(ctx, key, jsonStr, 7*24*time.Hour).Err()
	if err != nil {
		log.CacheLogger.Errorf("make rule cache failed: %v", err)
		return domain.CacheMakeError
	}
	return nil
}

func (r *RuleRepoImpl) Get(ctx context.Context, id int64) (*domain.Rule, error) {
	// 检查是否存在锁
	lockKey := fmt.Sprintf("lock:rule:%d", id)
	exist, err := r.cache.GetCache().Exists(ctx, lockKey).Result()
	if err != nil {
		log.CacheLogger.Errorf("get lock failed: %v", err)
		return nil, domain.CacheLockError
	}
	if exist == 1 {
		return nil, domain.CacheLocked
	}

	key := fmt.Sprintf("rule:%d", id)
	ruleJson, err := r.cache.GetCache().Get(ctx, key).Result()
	if err != nil {
		log.CacheLogger.Errorf("get rule cache failed: %v", err)
		return nil, domain.CacheGetError
	}

	var rule = &domain.Rule{}
	err = json.Unmarshal([]byte(ruleJson), rule)
	if err != nil {
		log.CacheLogger.Errorf("unmarshal rule cache failed: %v", err)
		return nil, domain.UnmarshalJsonError
	}

	if rule.Status == domain.StatusDeleted {
		return nil, domain.NotExist
	}
	if rule.Status == domain.StatusDisable {
		return nil, domain.Disable
	}

	return rule, nil
}

func (r *RuleRepoImpl) GetRuleUnscoped(ctx context.Context, id int64) (*domain.Rule, error) {
	key := fmt.Sprintf("rule:%d", id)

	ruleJson, err := r.cache.GetCache().Get(ctx, key).Result()
	if err != nil {
		log.CacheLogger.Errorf("get rule cache failed: %v", err)
		return nil, domain.DatabaseGetError
	}

	var rule = &domain.Rule{}
	err = json.Unmarshal([]byte(ruleJson), rule)
	if err != nil {
		log.CacheLogger.Errorf("unmarshal rule cache failed: %v", err)
		return nil, domain.UnmarshalJsonError
	}

	return rule, nil
}

// LockWrite 并发锁写入, 避免缓存击穿
func (r *RuleRepoImpl) LockWrite(ctx context.Context, id int64) error {
	key := fmt.Sprintf("rule:%d", id)
	lockKey := fmt.Sprintf("lock:rule:%d", id)
	lockValue := idgen.GenerateStringUid()

	locked, err := r.cache.GetLock(ctx, lockKey, lockValue)
	if err != nil {
		log.CacheLogger.Errorf("get lock failed: %v", err)
		return domain.CacheLockError
	}

	if locked {
		defer r.cache.ReleaseLock(ctx, lockKey, lockValue)

		ruleJson, err := r.cache.GetCache().Get(ctx, key).Result()
		if err == nil && len(ruleJson) > 0 {
			return nil
		}

		sqlStr := "select * from star_chat.public.rules where id = $1;"
		rule := &domain.Rule{}

		err = r.db.GetDB().GetContext(ctx, rule, sqlStr, id)
		if err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				// 降级处理, 设置缓存为空值, 并且设置12小时过期
				err := r.cache.GetCache().Set(ctx, key, "null", 12*time.Hour).Err()
				if err != nil {
					log.CacheLogger.Errorf("cache null value failed: %v", err)
					return domain.CacheError
				}
				return domain.NotExist
			}
			log.AppLogger.Errorf("rule repo get rule failed: %v", err)
			return domain.DatabaseGetError
		}

		makeCacheError := r.MakeRuleCache(ctx, id, rule)
		if makeCacheError != nil {
			log.CacheLogger.Errorf("make rule cache failed: %v", err)
			return domain.CacheMakeError
		}
	} else {
		ruleJson, err := r.cache.GetCache().Get(ctx, key).Result()
		if err == nil && len(ruleJson) > 0 {
			return nil
		} else {
			// 获取锁失败, 且缓存不存在, 返回通知 service 层重试
			return domain.CacheLockError
		}
	}

	return nil
}

func (r *RuleRepoImpl) GetAll(ctx context.Context) ([]*domain.Rule, error) {
	sqlStr := "select *  from star_chat.public.rules;"
	var rules []*domain.Rule

	err := r.db.GetDB().SelectContext(ctx, rules, sqlStr)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			log.SQLLogger.Errorf("get all rules failed: %v", err)
			return nil, domain.DatabaseGetError
		}
		log.AppLogger.Errorf("rule repo get all rules failed: %v", err)
		return nil, domain.DatabaseGetError
	}
	return nil, nil
}

func (r *RuleRepoImpl) Update(ctx context.Context, rule *domain.Rule) error {
	// 定义事务等级和是否只读
	opt := &sql.TxOptions{
		Isolation: sql.LevelDefault,
		ReadOnly:  false,
	}

	// 开启事务
	tx, err := r.db.GetDB().BeginTxx(ctx, opt)
	if err != nil {
		log.SQLLogger.Errorf("update rule begin tx failed: %v", err)
		return domain.DatabaseTXBeginError
	}

	selectStr := "select id, update_time from star_chat.public.rules where id = $1"
	updateStr := "update star_chat.public.rules set name = $1, description = $2, status = $3, conversation_type = $4, target_user_role = $5, permission = $6, messagehanding = $7, time_conditions = $8, trigger_conditions =  $9, action = $10, update_time = now() where id = $11 and update_time = $12;"

	lockMap := map[string]interface{}{
		"id":          nil,
		"update_time": nil,
	}
	err = tx.GetContext(ctx, lockMap, selectStr, rule.ID)
	if err != nil {
		txError := domain.DatabaseGetError
		defer func() {
			rollbackErr := tx.Rollback()
			if rollbackErr != nil {
				log.SQLLogger.Errorf("update rule rollback failed: %v", rollbackErr)
				txError = domain.DatabaseTXRollbackError
			}
		}()
		log.SQLLogger.Errorf("update rule lock failed: %v", err)
		return txError
	}

	result, err := tx.ExecContext(ctx, updateStr,
		rule.Name,
		rule.Description,
		rule.Status,
		rule.ConversationType,
		rule.TargetUserRole,
		rule.Permission,
		rule.MessageHanding,
		rule.TimeConditions,
		rule.TriggerConditions,
		rule.Action,
		lockMap["id"],
		lockMap["update_time"])
	if err != nil {
		log.SQLLogger.Errorf("update rule failed: %v", err)
		return domain.DatabaseUpdateError
	}

	rowsAffected, err := result.RowsAffected()
	if err != nil {
		log.SQLLogger.Errorf("get rows affected failed: %v", err)
		return domain.DatabaseError
	}

	if rowsAffected == 0 {
		// 乐观锁冲突：数据已被其他事务修改
		log.SQLLogger.Errorln("update rule failed due to optimistic lock conflict")
		return domain.DatabaseLocked
	}

	err = tx.Commit()
	if err != nil {
		txError := domain.DatabaseTXCommitError
		log.SQLLogger.Errorf("update rule commit failed: %v", err)
		defer func() {
			err = tx.Rollback()
			if err != nil {
				log.SQLLogger.Errorf("update rule rollback failed: %v", err)
				txError = domain.DatabaseTXRollbackError
			}
		}()
		return txError
	}

	log.MQLogger.Infof("update rule: %v success", rule.ID)
	return nil
}

//func (r *RuleRepoImpl) Update(ctx context.Context, rule *domain.Rule) error {
//	// 事务锁语句
//	lockStr := "select * from star_chat.public.rules where id = ? for update;"
//	// 修改语句
//	updateStr := "update star_chat.public.rules set name = ?, description = ?, status = ?, conversation_type = ?, target_user_role = ?, permission = ?, messagehanding = ?, time_conditions = ?, trigger_conditions = ?, action = ?, update_time = now() where id = ?;"
//
//	// 定义事务等级和是否只读
//	opt := &sql.TxOptions{
//		Isolation: sql.LevelDefault,
//		ReadOnly:  false,
//	}
//
//	// 开启事务
//	tx, err := r.db.GetDB().BeginTxx(ctx, opt)
//	if err != nil {
//		log.SQLLogger.Errorf("update rule begin tx failed: %v", err)
//		return errors.New("update rule begin tx failed")
//	}
//
//	// 获取事务锁
//	_, err = tx.ExecContext(ctx, lockStr, rule.ID)
//	if err != nil {
//		log.SQLLogger.Errorf("update rule lock failed: %v", err)
//		defer func()  {
//			err = tx.Rollback()
//			if err != nil {
//				log.SQLLogger.Errorf("update rule rollback failed: %v", err)
//			}
//		}()
//		return errors.New("update rule lock failed")
//	}
//
//	_, err = tx.ExecContext(ctx, updateStr,
//		rule.Name,
//		rule.Description,
//		rule.Status,
//		rule.ConversationType,
//		rule.TargetUserRole,
//		rule.Permission,
//		rule.MessageHanding,
//		rule.TimeConditions,
//		rule.TriggerConditions,
//		rule.Action,
//		rule.ID)
//	if err != nil {
//		log.SQLLogger.Errorf("update rule failed: %v", err)
//		return errors.New("update rule failed")
//	}
//
//	return nil
//}

func (r *RuleRepoImpl) Enable(ctx context.Context, id int64) error {
	// 定义事务等级和是否只读
	opt := &sql.TxOptions{
		Isolation: sql.LevelDefault,
		ReadOnly:  false,
	}

	// 开启事务
	tx, err := r.db.GetDB().BeginTxx(ctx, opt)
	if err != nil {
		log.SQLLogger.Errorf("update rule begin tx failed: %v", err)
		return domain.DatabaseTXBeginError
	}

	selectStr := "select id, update_time from star_chat.public.rules where id = $1"
	updateStr := "update star_chat.public.rules set status = 'enable', update_time = now() where id = $1 and  update_time = $2;"

	lockMap := map[string]interface{}{
		"id":          nil,
		"update_time": nil,
	}
	err = tx.GetContext(ctx, lockMap, selectStr, id)
	if err != nil {
		log.SQLLogger.Errorf("update rule lock failed: %v", err)
		return domain.DatabaseGetError
	}

	result, err := tx.ExecContext(ctx, updateStr, lockMap["id"], lockMap["update_time"])
	if err != nil {
		log.SQLLogger.Errorf("enable rule failed: %v", err)
		return domain.DatabaseUpdateError
	}

	rowsAffected, err := result.RowsAffected()
	if err != nil {
		log.SQLLogger.Errorf("get rows affected failed: %v", err)
		return domain.DatabaseError
	}

	if rowsAffected == 0 {
		log.SQLLogger.Errorln("enable rule failed due to optimistic lock conflict")
		return domain.DatabaseLocked
	}

	err = tx.Commit()
	if err != nil {
		txError := domain.DatabaseTXCommitError
		log.SQLLogger.Errorf("enable rule commit failed: %v", err)
		defer func() {
			err = tx.Rollback()
			if err != nil {
				log.SQLLogger.Errorf("enable rule rollback failed: %v", err)
				txError = domain.DatabaseTXRollbackError
			}
		}()
		return txError
	}

	log.SQLLogger.Infof("enable rule: %v success", id)
	return nil
}

func (r *RuleRepoImpl) EnableCache(ctx context.Context, id int64) error {
	cacheKey := fmt.Sprintf("rule:%d", id)
	lockKey := fmt.Sprintf("lock:rule:%d", id)
	lockValue := idgen.GenerateStringUid()
	expiration := 10 * time.Second

	ok, err := r.cache.GetCache().SetNX(ctx, lockKey, lockValue, expiration).Result()
	if err != nil {
		log.CacheLogger.Errorf("acquire lock failed: %v", err)
		return domain.CacheLockError
	}
	if !ok {
		log.CacheLogger.Errorln("distributed lock already held by another client")
		return domain.CacheLocked
	}

	defer func() {
		currentValue, lockErr := r.cache.GetCache().Get(ctx, lockKey).Result()
		if lockErr == nil && currentValue == lockValue {
			r.cache.GetCache().Del(ctx, lockKey)
		}
	}()

	// 获取原规则信息
	oldRuleJson, err := r.cache.GetCache().Get(ctx, cacheKey).Result()
	if err != nil {
		log.CacheLogger.Errorf("get rule from cache failed: %v", err)
		return domain.CacheGetError
	}
	rule := &domain.Rule{}
	err = json.Unmarshal([]byte(oldRuleJson), rule)
	if err != nil {
		log.CacheLogger.Errorf("unmarshal rule from cache failed: %v", err)
		return domain.UnmarshalJsonError
	}

	// 修改规则状态, 并重新写入覆盖原有缓存
	rule.Status = domain.StatusEnabled
	repoErr := r.MakeRuleCache(ctx, id, rule)

	return repoErr
}

func (r *RuleRepoImpl) Disable(ctx context.Context, id int64) error {
	lockStr := "select * from star_chat.public.rules where id = $1 for update;"
	updateStr := "update star_chat.public.rules set status = 'disabled', update_time = now() where id = $1;"

	// 定义事务等级和是否只读
	opt := &sql.TxOptions{
		Isolation: sql.LevelDefault,
		ReadOnly:  false,
	}

	// 开启事务
	tx, err := r.db.GetDB().BeginTxx(ctx, opt)
	if err != nil {
		log.SQLLogger.Errorf("diable rule begin tx failed: %v", err)
		return domain.DatabaseTXBeginError
	}

	// 获取事务锁
	_, err = tx.ExecContext(ctx, lockStr, id)
	if err != nil {
		txError := domain.DatabaseLockError
		log.SQLLogger.Errorf("diable rule lock failed: %v", err)
		defer func() {
			err = tx.Rollback()
			if err != nil {
				log.SQLLogger.Errorf("diable rule rollback failed: %v", err)
				txError = domain.DatabaseTXRollbackError
			}
		}()
		return txError
	}

	_, err = tx.ExecContext(ctx, updateStr, id)
	if err != nil {
		log.SQLLogger.Errorf("disable rule failed: %v", err)
		return domain.DatabaseUpdateError
	}

	err = tx.Commit()
	if err != nil {
		txError := domain.DatabaseTXCommitError
		log.SQLLogger.Errorf("diable rule commit failed: %v", err)
		defer func() {
			err = tx.Rollback()
			if err != nil {
				log.SQLLogger.Errorf("diable rule rollback failed: %v", err)
				txError = domain.DatabaseTXRollbackError
			}
		}()
		return txError
	}

	return nil
}

func (r *RuleRepoImpl) DisableCache(ctx context.Context, id int64) error {
	cacheKey := fmt.Sprintf("rule:%d", id)
	lockKey := fmt.Sprintf("lock:rule:%d", id)
	lockValue := idgen.GenerateStringUid()
	expiration := 10 * time.Second

	ok, err := r.cache.GetCache().SetNX(ctx, lockKey, lockValue, expiration).Result()
	if err != nil {
		log.CacheLogger.Errorf("acquire lock failed: %v", err)
		return domain.CacheLockError
	}
	if !ok {
		log.CacheLogger.Errorln("distributed lock already held by another client")
		return domain.CacheLocked
	}

	defer func() {
		currentValue, err := r.cache.GetCache().Get(ctx, lockKey).Result()
		if err == nil && currentValue == lockValue {
			r.cache.GetCache().Del(ctx, lockKey)
		}
	}()

	// 获取原规则信息
	oldRuleJson, err := r.cache.GetCache().Get(ctx, cacheKey).Result()
	if err != nil {
		log.CacheLogger.Errorf("get rule from cache failed: %v", err)
		return domain.CacheGetError
	}
	rule := &domain.Rule{}
	err = json.Unmarshal([]byte(oldRuleJson), rule)
	if err != nil {
		log.CacheLogger.Errorf("unmarshal rule from cache failed: %v", err)
		return domain.UnmarshalJsonError
	}

	// 修改规则状态, 并重新写入覆盖原有缓存
	rule.Status = domain.StatusDisable
	repoErr := r.MakeRuleCache(ctx, id, rule)

	return repoErr
}

func (r *RuleRepoImpl) Delete(ctx context.Context, id int64) error {
	lockStr := "select * from star_chat.public.rules where id = $1 for update;"
	deleteStr := "update star_chat.public.rules set status = 'deleted', delete_time = now() where id = $2;"

	// 定义事务等级和是否只读
	opt := &sql.TxOptions{
		Isolation: sql.LevelDefault,
		ReadOnly:  false,
	}

	// 开启事务
	tx, err := r.db.GetDB().BeginTxx(ctx, opt)
	if err != nil {
		log.SQLLogger.Errorf("update rule begin tx failed: %v", err)
		return domain.DatabaseTXBeginError
	}

	// 获取事务锁
	_, err = tx.ExecContext(ctx, lockStr, id)
	if err != nil {
		txError := domain.DatabaseLockError
		log.SQLLogger.Errorf("update rule lock failed: %v", err)
		defer func() {
			err = tx.Rollback()
			if err != nil {
				log.SQLLogger.Errorf("update rule rollback failed: %v", err)
				txError = domain.DatabaseTXRollbackError
			}
		}()
		return txError
	}

	_, err = tx.ExecContext(ctx, deleteStr, id)
	if err != nil {
		log.SQLLogger.Errorf("delete rule failed: %v", err)
		return domain.DatabaseDeleteError
	}

	err = tx.Commit()
	if err != nil {
		txError := domain.DatabaseTXCommitError
		log.SQLLogger.Errorf("delete rule commit failed: %v", err)
		defer func() {
			err = tx.Rollback()
			if err != nil {
				log.SQLLogger.Errorf("delete rule rollback failed: %v", err)
				txError = domain.DatabaseTXRollbackError
			}
		}()
		return txError
	}

	return nil
}

func (r *RuleRepoImpl) DeleteCache(ctx context.Context, id int64) error {
	lockKey := fmt.Sprintf("lock:rule:%d", id)
	lockValue := idgen.GenerateStringUid()
	expiration := 10 * time.Second

	ok, err := r.cache.GetCache().SetNX(ctx, lockKey, lockValue, expiration).Result()
	if err != nil {
		log.CacheLogger.Errorf("acquire lock failed: %v", err)
		return domain.CacheLockError
	}
	if !ok {
		log.CacheLogger.Errorln("distributed lock already held by another client")
		return domain.CacheLocked
	}

	defer func() {
		currentValue, err := r.cache.GetCache().Get(ctx, lockKey).Result()
		if err == nil && currentValue == lockValue {
			r.cache.GetCache().Del(ctx, lockKey)
		}
	}()

	cacheKey := fmt.Sprintf("rule:%d", id)
	// 获取原规则信息
	oldRuleJson, err := r.cache.GetCache().Get(ctx, cacheKey).Result()
	if err != nil {
		log.CacheLogger.Errorf("get rule from cache failed: %v", err)
		return domain.CacheGetError
	}
	rule := &domain.Rule{}
	err = json.Unmarshal([]byte(oldRuleJson), rule)
	if err != nil {
		log.CacheLogger.Errorf("unmarshal rule from cache failed: %v", err)
		return domain.UnmarshalJsonError
	}

	// 修改规则状态, 并重新写入覆盖原有缓存
	rule.Status = domain.StatusDeleted
	repoErr := r.MakeRuleCache(ctx, id, rule)

	return repoErr
}

// IsExist  通过id查询是否存在, 如果存在返回true, 不存在返回false
func (r *RuleRepoImpl) IsExist(ctx context.Context, id int64) (bool, error) {
	// 先从缓存中查询
	key := fmt.Sprintf("rule:%d", id)
	cacheExist, err := r.cache.GetCache().Exists(ctx, key).Result()
	if err != nil {
		log.CacheLogger.Errorf("check rule exist failed: %v", err)
		return false, domain.CacheGetError
	}
	// 缓存中存在, 直接返回结果
	if cacheExist == 1 {
		result, err := r.cache.GetCache().Get(ctx, key).Result()
		if err != nil {
			log.CacheLogger.Errorf("check rule exist failed: %v", err)
			return false, domain.CacheGetError
		}
		// 如果缓存中存在, 但是值是nil, 说明规则不存在数据库中, 进行了降级处理, 也返回false
		if result == domain.CacheNullValue {
			return false, nil
		} else {
			// 缓存中存在, 且值不为nil, 说明规则存在, 直接返回true
			return true, nil
		}
	}

	var dbExist bool
	existStr := "select exists(select 1 from star_chat.public.rules where id = $1);"

	err = r.db.GetDB().GetContext(ctx, &dbExist, existStr, id)
	if err != nil {
		log.SQLLogger.Errorf("check rule exist failed: %v", err)
		return false, domain.DatabaseGetError
	}
	// 若数据库中存在, 则将结果写入缓存
	if dbExist == true {
		getStr := "select * from star_chat.public.rules where id = $1;"
		rule := &domain.Rule{}

		err = r.db.GetDB().GetContext(ctx, rule, getStr, id)
		if err != nil {
			log.SQLLogger.Errorf("get rule failed: %v", err)
			return false, domain.DatabaseGetError
		}

		makeCacheErr := r.MakeRuleCache(ctx, id, rule)
		if makeCacheErr != nil {
			return false, makeCacheErr
		}
		return true, nil
	}

	// 若数据库中不存在, 则将结果设为空, 降级处理, 写入缓存, 防止缓存穿透
	makeCacheError := r.cache.GetCache().Set(ctx, key, domain.CacheNullValue, 24*time.Hour).Err()
	if makeCacheError != nil {
		log.CacheLogger.Errorf("set rule null cache failed: %v", err)
		return false, domain.CacheMakeError
	}
	return false, nil
}

// IsExistByName  通过name查询是否存在, 如果存在返回true, 不存在返回false, 调用次数较少, 后期经过评估, 考虑是否需要建立缓存
func (r *RuleRepoImpl) IsExistByName(ctx context.Context, name string) (bool, error) {
	sqlStr := "select exists(select 1 from star_chat.public.rules where name = $1);"

	var exist bool
	err := r.db.GetDB().GetContext(ctx, exist, sqlStr, name)
	if err != nil {
		log.SQLLogger.Errorf("check rule exist failed: %v", err)
		return false, domain.DatabaseGetError
	}
	return exist, nil
}
